package com.maben.mqtt.consumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 自己的消费者处理类
 */
@Slf4j(topic = "m.MyConsumerHandler")
public class MyConsumerHandler implements MessageHandler {
    /**
     * 序号 检查接收顺序
     */
    private static final AtomicLong NUM = new AtomicLong(-1);
    @Override
    public void handleMessage(Message<?> message) throws MessagingException {
        final byte[] payload = (byte[]) message.getPayload();
        final String payloadStr = new String(payload, StandardCharsets.UTF_8);
        final long aLong = Long.parseLong(payloadStr);
        if (NUM.get()==-1){
            NUM.set(aLong);
        }else {
            final long cacheNum = NUM.addAndGet(1);
            if (cacheNum!=aLong){
                log.error("缓存中序号为：{}，接收到序号为：{}",cacheNum,aLong);
            }
        }
        log.info("message:" + aLong);
//        System.out.println("header:" + JSON.toJSONString(message.getHeaders()));
//        String topic = (String) message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC);
//        System.out.println("topic:" + topic);
    }
}
